package com.clougence.cloudcanal.dataprocess.datatransform;

import com.clougence.cloudcanal.dataprocess.datatransform.common.DateUtils;
import com.clougence.cloudcanal.dataprocess.datatransform.common.GlanceRecordConstants;
import com.clougence.cloudcanal.dataprocess.datatransform.common.JsonUtil;
import com.clougence.cloudcanal.dataprocess.datatransform.converter.DwdGlanceRecordConverter;
import com.clougence.cloudcanal.dataprocess.datatransform.model.GlanceRecordBO;
import com.clougence.cloudcanal.dataprocess.datatransform.model.OdsGlanceEventRecordBO;
import com.clougence.cloudcanal.sdk.api.CloudCanalProcessorV2;
import com.clougence.cloudcanal.sdk.api.JavaDsType;
import com.clougence.cloudcanal.sdk.api.ProcessorContext;
import com.clougence.cloudcanal.sdk.api.modelv2.CustomData;
import com.clougence.cloudcanal.sdk.api.modelv2.CustomFieldV2;
import com.clougence.cloudcanal.sdk.api.modelv2.CustomRecordV2;
import com.clougence.cloudcanal.sdk.api.modelv2.SchemaInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import javax.sql.DataSource;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import static com.clougence.cloudcanal.dataprocess.datatransform.common.GlanceRecordConstants.*;
import static com.clougence.cloudcanal.dataprocess.datatransform.converter.DwdGlanceRecordConverter.buildCustomRecord;

/**
 * @author yongSen.wang
 * @date 2022/03/09 15:32
 */
@Slf4j
public class MySqlToClickhouseGlanceEvent implements CloudCanalProcessorV2 {

    @Override
    public void start(ProcessorContext processorContext) {
    }

    @Override
    public List<CustomData> process(CustomData data) {
        List<CustomData> re = new ArrayList<>();
        handleFactTable(data);
        re.add(data);

        return re;
    }

    protected void handleFactTable(CustomData data) {
        for (CustomRecordV2 recordV2 : data.getRecords()) {
            CustomFieldV2 f = null;
            CustomFieldV2 ipField = null;
            CustomFieldV2 timeField = null;
            switch (data.getEventType()) {
                case INSERT:
                case UPDATE:
                    f = recordV2.getAfterColumnMap().get(SOURCE_DATA);
                    ipField = recordV2.getAfterColumnMap().get(IP);
                    timeField = recordV2.getAfterColumnMap().get(CREATE_TIME);
                    break;
                case DELETE:
                    break;
                default:
                    throw new IllegalArgumentException(
                        "unsupported event type:" + data.getEventType());
            }
            // SOURCE_DATA为空则删除此条数据
            if (f == null) {
                continue;
            }
            String source = String.valueOf(f.getValue());
            if (StringUtils.isBlank(source)) {
                continue;
            }
            GlanceRecordBO bo = DwdGlanceRecordConverter.build(source, ipField, timeField);

            buildCustomRecord(recordV2, bo);
        }

    }

    @Override
    public void stop() {

    }
}
